package com.yanxu.source;

import com.yanxu.domain.Event;
import com.yanxu.domain.Event2;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025/7/27-2025
 * @Description: 自定义 source
 */
@Slf4j
public class CustomSource2 implements SourceFunction<Event2> {
    boolean running = true;

    @Override
    public void run(SourceContext<Event2> ctx) throws Exception {
        // 创建Random实例，用于生成随机数
        Random random = new Random();
        // 定义用户数组和访问URL数组
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};

        // 当running为true时，持续生成事件
        while (running) {
            // 随机选择一个用户和URL，结合当前时间戳，创建一个新的Event对象并收集
            ctx.collect(new Event2(
                    users[random.nextInt(users.length)],
                    urls[random.nextInt(urls.length)],
                    new Random().nextInt(1000),
                    Calendar.getInstance().getTimeInMillis()
            ));
            log.error("custom source .....");
            // 为了便于观察和处理，每生成一个数据后，暂停50毫秒
            Thread.sleep(50);
        }
    }

    @Override
    public void cancel() {
        running=false;
    }
}